package com.demo.rocketmq.listener;

import cn.hutool.core.util.IdUtil;
import com.demo.rocketmq.task.DemoTask;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageClientExt;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.FutureTask;
import java.util.stream.Collectors;

/**
 * @author wangfengchen
 */
@Component
@Slf4j
public class DemoConcurrentConsumeListener implements MessageListenerConcurrently {

//    private int minThreads = 5;
//    private int maxThreads = 100;
//    private int MAX_QUEUE_SIZE = 2000;
//    private ExecutorService executor = new ThreadPoolExecutor(minThreads,
//            maxThreads, 0L, TimeUnit.MILLISECONDS,
//            new ArrayBlockingQueue<Runnable>(MAX_QUEUE_SIZE),
//            new ThreadPoolExecutor.CallerRunsPolicy());

  @Autowired
  private ThreadPoolTaskExecutor threadPoolTaskExecutor;


  /**
   * 默认list里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息<br/>
   * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
   */
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
    String requestId = IdUtil.getSnowflakeNextIdStr();
    log.info("requestId = {}，【ConsumeConcurrentlyStatus】消费者收到消息，msgSize = {}", requestId, list.size());
    if (CollectionUtils.isEmpty(list)) {
      log.error("requestId = {}，消费者接受到的消息为空，不处理，直接返回成功", requestId);
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    List<String> msgKeyList = list.stream().map(Message::getKeys).collect(Collectors.toList());
    log.info("requestId = {}，消息key集合，msgKeyList = {}", requestId, msgKeyList);
    boolean flag = true;
    for (MessageExt messageExt : list) {
//      String msgId;
//      // 此处注意当类型为MessageClientExt时，getMsgId()不是真的msgId，而是UniqueKey，所以需要调用getOffsetMsgId()方法
//      if (messageExt instanceof MessageClientExt) {
//        MessageClientExt ext = (MessageClientExt) messageExt;
//        msgId = ext.getOffsetMsgId();
//      } else {
//        msgId = messageExt.getMsgId();
//      }
      String msg = new String(messageExt.getBody());
      String topic = messageExt.getTopic();
      String tags = messageExt.getTags();
      String key = messageExt.getKeys();
      int reconsume = messageExt.getReconsumeTimes();
      // 如果达到重试次数不成功，则记录，人工补偿
      if (reconsume >= 16) {
        log.error("DemoSequenceConsumerListener消费重试次数达到" + reconsume + "次，不再重试，返回成功，人工补偿");
        // TODO 人工补偿，达到最大次数后从redis中查询出成功的删除掉，将消费失败的存入redis进行人工补偿，补偿后再删除
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
      log.info("requestId = {}，消息内容，key = {}, msg = {}，topic = {}，tags = {}", requestId, key, msg, topic, tags);
      try {
        // TODO 判断该消息是否重复消费，可将成功消费的消息存入redis并设置过期时间要大于重试次数的时间间隔，同时continue
        // TODO 处理对应的业务逻辑，使用线程池和异步任务处理
        DemoTask demoTask = new DemoTask(requestId, key, msg);
        FutureTask<Boolean> result = new FutureTask<>(demoTask);
        threadPoolTaskExecutor.execute(result);
        Boolean isTrue = result.get();
        log.info("requestId = {}，异步任务返回结果{}，key = {}", requestId, isTrue ? "成功" : "失败", key);
        // 如果有一条失败则将整体消息标记为失败
        if (!isTrue) {
          flag = false;
        }
      } catch (Exception e) {
        log.error("requestId = {}，消费者异常，key = {}，e = {}", requestId, key, ExceptionUtils.getStackTrace(e));
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
      }
    }
    // 如果全部成功则返回成功，有一条失败就重新消费
    if (flag) {
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  }
}
